package com.edata.bigdata.nebulasource

import com.vesoft.nebula.client.graph.NebulaPoolConfig
import com.vesoft.nebula.client.graph.data.{HostAddress, ResultSet}
import com.vesoft.nebula.client.graph.net.{NebulaPool, Session}

import scala.collection.JavaConverters._

class NebulaConnector(addrs: String,
                      user: String,
                      password: String,
                      enableSSL: Boolean = false,
                      timeout: Int) extends AutoCloseable with Serializable {

  @transient val nebulaPoolConfig = new NebulaPoolConfig
  @transient val pool: NebulaPool = new NebulaPool //创建pool后会使程序处于运行状态不会结束

  var addresses = addrs.split(",").map(ipport => {
    val ip_port = ipport.split(":")
    new HostAddress(ip_port(0), ip_port(1).toInt)
  }).toList
  nebulaPoolConfig.setMaxConnSize(100)
  nebulaPoolConfig.setTimeout(timeout)

  if (enableSSL) {
  }
  val randAddr = scala.util.Random.shuffle(addresses)
  pool.init(randAddr.asJava, nebulaPoolConfig)
  lazy val session: Session = pool.getSession(user, password, true)

  def submit(ngql: String): ResultSet = {
    var rs: ResultSet = null
    try {
      rs = session.execute(ngql)
    } catch {
      case e: Exception => e.printStackTrace()
    } finally {
      session.release()
      session.close()
    }
    rs

  }
  override def close(): Unit = {
    session.release()
    session.close()
    pool.close()

  }

}
